// 在 com.bw.gmall.realtime.app.dim 包中创建 TableProcessFunction 类
package com.bw.gmall.realtime.app.dim;

import com.bw.gmall.realtime.bean.TableProcess;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import com.alibaba.fastjson.JSONObject;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        // 处理业务数据
        String tableName = jsonObj.getString("table");
        TableProcess tableProcess = broadcastState.get(tableName);

        if (tableProcess != null) {
            // 添加 sinkTable 信息到数据中
            jsonObj.put("sinkTable", tableProcess.getSinkTable());
            out.collect(jsonObj);
        }
    }

    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        // 处理配置数据并更新广播状态
        JSONObject jsonObj = JSONObject.parseObject(value);
        TableProcess tableProcess = JSONObject.parseObject(jsonObj.getString("after"), TableProcess.class);

        // 更新广播状态
        ctx.getBroadcastState(mapStateDescriptor).put(tableProcess.getSourceTable(), tableProcess);
    }
}
